package client

import (
	"shen/gotask/pkg/gotask"
	"shen/gotask/rocketmq/config"
	"shen/gotask/rocketmq/driver/rocketmq"
	"shen/gotask/rocketmq/logs"
	"sync"
	"time"
)

type Client struct {
	Config                     config.Model
	DriverCloseChannel         chan bool
	Logs                       *logs.Logs
	RunConsumer                map[string]rocketmq.RunConsumerCloseChan
	NeedleCloseConsumerChannel chan string
	NeedleRunConsumerChannel   chan rocketmq.ConsumerResult
	InitCount                  int
}

func closeChannelIfNotClosed(ch rocketmq.RunConsumerCloseChan) {
	select {
	case _, ok := <-ch:
		if !ok {
			return
		}
	default:
		close(ch)
	}
}

func (c *Client) Run(wg *sync.WaitGroup) {
	defer wg.Done()
	defer func() {
		if err := recover(); err != nil {
			c.Logs.Error("消费者运行失败！")
		}
	}()
	defer close(c.DriverCloseChannel)
	c.RunConsumer = make(map[string]rocketmq.RunConsumerCloseChan)
	c.NeedleCloseConsumerChannel = make(chan string, 1024)
	defer close(c.NeedleCloseConsumerChannel)
	c.NeedleRunConsumerChannel = make(chan rocketmq.ConsumerResult, 1024)
	defer close(c.NeedleRunConsumerChannel)

	var wgC sync.WaitGroup
	wgC.Add(1)
	//第一个，需要启动一个协程去定时获取php的消息队列
	go c.initConsumer(&wgC)

	wgC.Add(1)
	go c.startConsumer(&wgC)

	wgC.Add(1)
	go func() {
		for {
			select {
			case <-c.DriverCloseChannel:
				return
			default:
				needleCloseConsumer := <-c.NeedleCloseConsumerChannel
				if data, ok := c.RunConsumer[needleCloseConsumer]; ok {
					closeChannelIfNotClosed(data)
					delete(c.RunConsumer, needleCloseConsumer)
				}
			}
		}
	}()

	wgC.Wait()
}

func (c *Client) startConsumer(wgC *sync.WaitGroup) {
	defer wgC.Done()
	defer func() {
		if err := recover(); err != nil {
			c.Logs.Error("消费者运行失败，错误信息：", err)
			close(c.DriverCloseChannel)
		}
	}()
	for {
		select {
		case <-c.DriverCloseChannel:
			return
		default:
			consumerResult := <-c.NeedleRunConsumerChannel
			name := consumerResult.Class + "::" + consumerResult.Method
			if _, ok := c.RunConsumer[name]; !ok {
				consumerClose := make(chan bool)
				c.RunConsumer[name] = consumerClose
				consumer := &rocketmq.Consumer{}
				go consumer.DoConsumer(consumerResult, consumerClose, c.NeedleCloseConsumerChannel)
			} else {
				time.Sleep(time.Second * 5)
			}
		}
	}
}

func (c *Client) initConsumer(wgC *sync.WaitGroup) {
	defer wgC.Done()
	defer func() {
		if err := recover(); err != nil {
			c.Logs.Error("初始化消费者失败，错误信息：", err)
			close(c.DriverCloseChannel)
		}
	}()
	for {
		select {
		case <-c.DriverCloseChannel:
			return
		default:
			c.createConsumer()
			time.Sleep(time.Second * 3)
		}
	}
}

func (c *Client) createConsumer() {
	client, errC := gotask.NewPhpClient()
	if errC != nil {
		c.InitCount++
		if c.InitCount >= 3 {
			for name, _ := range c.RunConsumer {
				c.NeedleCloseConsumerChannel <- name
			}
			c.InitCount = 0
		}
		return
	}
	defer client.Close()

	param := c.getIsRun()
	var res rocketmq.Result
	err := client.Call("Shen\\Rocketmq5\\Service\\ConfigService::getConfig", param, &res)
	if err != nil {
		c.Logs.Error("初始化超时了：", err)
	}

	if res.Result {
		c.InitCount = 0
		for _, result := range res.NeedleRun {
			c.NeedleRunConsumerChannel <- result
		}
		for _, result := range res.NeedleClose {
			c.NeedleCloseConsumerChannel <- result
		}
	} else {
		c.InitCount++
		if c.InitCount >= 3 {
			for name, _ := range c.RunConsumer {
				c.NeedleCloseConsumerChannel <- name
			}
			c.InitCount = 0
		}
	}
}

func (c *Client) getIsRun() rocketmq.ConsumerParams {
	var param = rocketmq.ConsumerParams{
		Name: []string{},
	}
	for name, _ := range c.RunConsumer {
		param.Name = append(param.Name, name)
	}
	return param
}
